package com.xxx.flink.sink;

import com.xxx.flink.customsource.CustomSource;
import com.xxx.flink.pojo.Event;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Sink到elasticsearch
 */
public class SinkToElasticsearch {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据
        DataStreamSource<Event> sourceStream = env.addSource(new CustomSource());

        // sink到elasticsearch
        HttpHost httpHost = new HttpHost("localhost", 9200);
        List<HttpHost> httpList = new ArrayList<>();
        httpList.add(httpHost);

        ElasticsearchSinkFunction<Event> elasticsearchSinkFunction = new ElasticsearchSinkFunction<Event>() {

            @Override
            public void open() throws Exception {
                System.out.println("open()");
            }

            @Override
            public void close() throws Exception {
                System.out.println("close()");
            }

            @Override
            public void process(Event event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
                Map<String, String> map = new HashMap<>();
                map.put(event.getName(), event.getUrl());

                IndexRequest indexRequest = Requests.indexRequest().index("click").source(map);
                requestIndexer.add(indexRequest);
            }
        };

        sourceStream.addSink(new ElasticsearchSink.Builder<>(httpList, elasticsearchSinkFunction).build());

        env.execute("elasticsearch");
    }


}
